package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;


public class DimUtil {
    //从phoenix查询数据，优化，当存在条件的时候
    //TODO 1 直接从phoenix中查询数据 也就是说只定义了查询的sql语句，只加入了条件判断,将此查询sql传给之前的Phoex工具类中方法去查询而已
    public static  JSONObject getDimInfoNoCache(String tableName, Tuple2<String,String>... columnNameAndValues){
        //自定义查询sql,Tuple2都是条件,eg:   ID='14'   TM_NAME='测试插入'
        String dimsql="select * from "+tableName+" where ";
        for(int i = 0;i < columnNameAndValues.length;i++){
            //每次遍历tuple2时候，都获取他的 属性和值,也就是 id, 14 :  tm_name,测试插入，这些数据
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            //获取列名
            String columnName = columnNameAndValue.f0;
            //列名对应的属性
            String columnValue=columnNameAndValue.f1;
            if(i>=1){
                //大于1，说明多个判断条件，拼接and到sql
                dimsql+=" and ";
            }
                dimsql+=columnName+"='"+columnValue+"'";

        }
        System.out.println("查询的sql语句为:"+dimsql);


        JSONObject dimInfoJsonObj=null;
        //通过之前封装的连接查询对象和方法，将sql语句放进方法中
        List<JSONObject> dimInfoList = PhoenixUtil.queryList(dimsql, JSONObject.class);//返回一个查询的多条的数据集合
        //因为有查询条件，所以一般是查询出一条，集合中只有一条数据
        if(dimInfoList != null && dimInfoList.size()>0){

            dimInfoJsonObj = dimInfoList.get(0);
        }else {
            System.out.println("没有满足条件的维度数据,查询语句为:"+dimsql);
        }

        return dimInfoJsonObj;
    }


    //************************************************************************************************************

    //TODO 2 从缓存中读取数据通过phoenix查询数据过于缓慢，所以可以使用旁路缓存，将查询的数据写入到redis内存数据库中，
    //TODO 可以设置内存时间为一天，如果查询数据，如果缓存中有，直接查缓存，没有就查phoenix再写入redis里

    /*
       旁路缓存:先从缓存Redis中查询维度数据，如果Redis中存在，那么直接返回；
           如果Redis中不存在，再到Phoenix中查询，并且将查询到的维度数据放到Redis中缓存起来
       Redis
           type:   String
           key:    dim:维度表名:14_cc
           value:  json格式字符串
           ttl:    1day
   */

    public static JSONObject getDimInfoCache(String tableName, String id) {
        return getDimInfoCache(tableName,Tuple2.of("id",id));
    }


    public static  JSONObject getDimInfoCache(String tableName, Tuple2<String,String>... columnNameAndValues){
        //自定义查询sql,Tuple2都是条件,eg:   ID='14'   TM_NAME='测试插入'
        String dimsql="select * from "+tableName+" where ";

        //定义操作redis的key.需要id的值拼接name的值 key-> dim:BASE_TRADEMARK:14_测试插入
        String redisKey="dim:"+tableName.toLowerCase()+":";
        for(int i = 0;i < columnNameAndValues.length;i++){
            //每次遍历tuple2时候，都获取他的 属性和值,也就是 id, 14 :  tm_name,测试插入，这些数据
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            //获取列名
            String columnName = columnNameAndValue.f0;
            //列名对应的属性
            String columnValue=columnNameAndValue.f1;
            if(i>=1){
                //大于1，说明多个判断条件，拼接and到sql
                dimsql+=" and ";
                redisKey+="_";
            }
            dimsql+=columnName+"='"+columnValue+"'";
            //dim:BASE_TRADEMARK:14_测试插入
            redisKey+=columnValue;

        }
        System.out.println("查询的sql语句为:"+dimsql);


        //根据rediskey到redis中查询维度数据
        Jedis jedis=null;
        String dimJsonStr=null;
        JSONObject dimInfoJsonObj=null;

        try {
            jedis = RedisUtil.getJedis();
            dimJsonStr= jedis.get(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("从redis中生成缓存失败:"+redisKey);
        }

        //判断是否从redis缓存中读取到数据
        if(dimJsonStr !=null && dimJsonStr.length()>0){
            dimInfoJsonObj = JSONObject.parseObject(dimJsonStr);
        }else {
            System.out.println("在缓存中没有查到数据:"+dimsql);
            //再缓存中没有查到数据，再从phoenix中查找，然后再写入redis
            List<JSONObject> dimInfoList = PhoenixUtil.queryList(dimsql, JSONObject.class);
            if(dimInfoList !=null && dimInfoList.size()>0){
                dimInfoJsonObj = dimInfoList.get(0);
                //将查询的结果写入redis中
                if(jedis !=null){
                    jedis.setex(redisKey,3600*24,dimInfoJsonObj.toString());
                }
            }else {
                System.out.println("没有查询到维度数据:" + dimsql);
            }
        }


        //通过之前封装的连接查询对象和方法，将sql语句放进方法中
        List<JSONObject> dimInfoList = PhoenixUtil.queryList(dimsql, JSONObject.class);//返回一个查询的多条的数据集合
        //因为有查询条件，所以一般是查询出一条，集合中只有一条数据
        if(dimInfoList != null && dimInfoList.size()>0){

            dimInfoJsonObj = dimInfoList.get(0);
        }else {
            System.out.println("没有满足条件的维度数据,查询语句为:"+dimsql);
        }

        //关闭连接
        if(jedis!=null){
            jedis.close();
        }
        return dimInfoJsonObj;


    }

    //失效的缓存,进行关闭
    public static  void  deleteCache(String tableName,String id){
        //定义rediskey
        String rediskey="dim:"+tableName.toLowerCase()+":"+id;

        try {
            Jedis jedis = RedisUtil.getJedis();
            jedis.del(rediskey);
            jedis.close();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("删除redis中的缓存失败");
        }

    }

    public static void main(String[] args) {
        //调用非缓存
        //JSONObject dimInfoNoCache = getDimInfoNoCache("BASE_TRADEMARK", Tuple2.of("id", "14"),Tuple2.of("tm_name", "测试插入"));
        //调用缓存
        JSONObject dimInfoCache = getDimInfoCache("BASE_TRADEMARK", "14");
        System.out.println(dimInfoCache);
    }

}
